package com.ruoyi.project.business.controlcenter.iot.mqtt.handler;//package com.yunwow.apiserver.domain.iot.mqtt.handler;
//
//import cn.hutool.core.util.ObjectUtil;
//import cn.hutool.core.util.StrUtil;
//import com.alibaba.fastjson.JSON;
//import com.yunwow.apiserver.domain.iot.config.MqttProperties;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.messaging.Message;
//import org.springframework.messaging.MessageHandler;
//import org.springframework.messaging.MessagingException;
//
//import javax.annotation.Resource;
//import java.util.concurrent.ExecutorService;
//
//public class MqttReceiveHander implements MessageHandler {
//    @Autowired
//    private MqttProperties mqttProperties;
//
//    @Resource
//    private MqttMessageRouter mqttMessageRouter;
//    @Resource
//    private ExecutorService executorService;
//
//    @Override
//    public void handleMessage(Message<?> message) throws MessagingException {
//        MessagePayload messagePayload = JSON.parseObject(message.getPayload().toString(), MessagePayload.class);
//
//        // message 预处理(处理空字段
//        // 当类型为空时，使用默认类型
//        // 当消息载荷为空时，使用其 String 内容
//        if (ObjectUtil.isNull(messagePayload)) {
//            messagePayload = new MessagePayload(message.getPayload(), MessageType.DEFAULT);
//        } else {
//            if (StrUtil.isEmpty(messagePayload.getPayload())) messagePayload.setPayload(message.getPayload().toString());
//            if (ObjectUtil.isNull(messagePayload.getMessageType())) messagePayload.setMessageType(MessageType.DEFAULT);
//        }
//
//        MessageHeader messageHeader = new MessageHeader(message.getHeaders());
//        MqttMessage mqttMessage = new MqttMessage(messageHeader, messagePayload);
//
//        // Judge whether to process data in batch according to the data transmission speed
//        if (MqttCacheContainer.messageSpeed.get() < mqttProperties.getBatchSpeed()) {
//            MqttMessage finalMqttMessage = mqttMessage;
//            executorService.execute(() -> {
//                // Receive single mqtt message
//                mqttMessageRouter.messageRoute(finalMqttMessage);
//            });
//        } else {
//            // Save point value to schedule
//            MqttCacheContainer.messageLock.writeLock().lock();
//            MqttCacheContainer.mqttMessages.add(mqttMessage);
//            MqttCacheContainer.messageLock.writeLock().unlock();
//        }
//
//    }
//}
